package com.lagou.rpc.consumer;

import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@SpringBootApplication
public class LgRpcConsumerApplication implements CommandLineRunner {

    CuratorFramework client = null;
    ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

    public static void main(String[] args) {
        SpringApplication.run(LgRpcConsumerApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("base")
                .build();
        client.start();
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/lg-provider", true);
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);

        List<ChildData> childDataList = pathChildrenCache.getCurrentData();
        for (ChildData childData : childDataList) {
            String[] data = new String(childData.getData()).split(":");
            RpcClient rpcClient = new RpcClient(data[0], Integer.parseInt(data[1]));
            rpcClient.setPath(childData.getPath());
            RpcClientProxy.list.add(rpcClient);
        }

        // 添加事件监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                // 通过判断event type的方式来实现不同事件的触发
                if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {  // 添加子节点时触发
                    String[] data = new String(event.getData().getData()).split(":");
                    RpcClient rpcClient = new RpcClient(data[0], Integer.parseInt(data[1]));
                    rpcClient.setPath(event.getData().getPath());
                    RpcClientProxy.list.add(rpcClient);
                } else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {  // 删除子节点时触发
                    String[] data = new String(event.getData().getData()).split(":");
                    Iterator<RpcClient> iterator = RpcClientProxy.list.iterator();
                    while(iterator.hasNext()) {
                        RpcClient rpcClient = iterator.next();
                        if(rpcClient.getIp().equals(data[0]) &&
                            rpcClient.getPort() == Integer.parseInt(data[1])) {
                            iterator.remove();
                        }
                    }
                }
            }
        });

        service.scheduleAtFixedRate(() -> {
            long currentTime = System.currentTimeMillis();
            for (int i=0;i<RpcClientProxy.list.size();i++) {
                RpcClient rpcClient = RpcClientProxy.list.get(i);
                long responseTime = rpcClient.getLastResponseTime();
                if((currentTime - responseTime) > 5000) {
                    try {
                        /*String zkValue = new String(client.getData().forPath(rpcClient.getPath()));
                        zkValue = zkValue.substring(zkValue.indexOf("-"));
                        client.setData().forPath(key, zkValue.getBytes());*/
                        System.out.println(rpcClient.getPath() + "响应时间清零，当前时间为：" + System.currentTimeMillis());
                        rpcClient.setCostTime(0L);
                        RpcClientProxy.list.set(i, rpcClient);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }, 0, 5, TimeUnit.SECONDS);

    }
}
